package com.lsx143.wordcount.day2;

import com.lsx143.wordcount.bean.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class RandomWaterSensor implements SourceFunction<WaterSensor>, Serializable {
    private boolean isRun = true;
    private RUN_MODE run_mode;
    private int speed = 1;
    private int index = 1;
    private String path = "input/sensor.txt";
    private List<WaterSensor> sensors = new ArrayList<>();

    {
        try {
            BufferedReader bf = new BufferedReader(new FileReader(path));
            String line;
            while ((line = bf.readLine()) != null) {
                String[] split = line.split(",");
                System.out.println(line);
                sensors.add(
                        new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        ));
            }
            System.out.println("集合长度：" + sensors.size());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public enum RUN_MODE {
        RANDOM,
        INCREASE,
        BY_FILE
    }

    public RandomWaterSensor(RUN_MODE run_mode) {
        this.run_mode = run_mode;
    }

    @Override
    public void run(SourceContext<WaterSensor> ctx) throws Exception {
        long counter = 0;
        int cnt_index = 0;
        while (isRun) {
            WaterSensor waterSensor = null;
            if (run_mode == RUN_MODE.RANDOM) {
                Random random = new Random();
                int water = random.nextInt(50);
                int index = random.nextInt(getIndex());
                waterSensor = new WaterSensor("Sensor_" + index, System.currentTimeMillis(), water);
            } else if (run_mode == RUN_MODE.INCREASE) {
                counter += 1000;
                waterSensor = new WaterSensor("Sensor_1:", counter, (int) counter / 100);
            } else if (run_mode == RUN_MODE.BY_FILE) {
                if (cnt_index >= sensors.size()) {
                    cnt_index = 0;
                }
                waterSensor = sensors.get(cnt_index++);
            }
            ctx.collect(waterSensor);
            Thread.sleep(1000 / speed);
        }
    }

    @Override
    public void cancel() {
        isRun = false;
    }

}
